package com.joymeter.data.service;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.joymeter.data.mapper.VlinkerMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * @ClassName VlinkerService
 * 微领地增量同步
 * @Description TODO
 * @Author liang
 * @Date 2018/10/15 10:51
 * @Version 1.0
 **/
@Service
public class VlinkerService {

    private static final Logger logger = Logger.getLogger(VlinkerService.class.getName());

    @Autowired
    private VlinkerMapper mapper;
    private final static String ip = "192.168.25.38";
    private final static int port = 11111;

    public void connector(){
        logger.log(Level.INFO, "===========创建连接======= " );

        // 创建链接
        CanalConnector connector = CanalConnectors.
                newSingleConnector(new InetSocketAddress(ip, port), "example", "", "");
        int batchSize = 1000;
        int emptyCount = 0;
        try {
            connector.connect();
            connector.subscribe(".*\\..*");
            connector.rollback();
            int totalEmptyCount = 12000;
            while (true) {
                Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    emptyCount++;
                    //System.out.println("empty count : " + emptyCount);
                    //logger.log(Level.INFO, "empty count : " + emptyCount);


                } else {
                    emptyCount = 0;
                    // System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);
                    try {
                        printEntry(message.getEntries());
                    }catch (Exception e){
                        System.out.println(e);
                        logger.log(Level.SEVERE, ""+e);

                    }
                 }

                connector.ack(batchId); // 提交确认
                // connector.rollback(batchId); // 处理失败, 回滚数据
            }

            //System.out.println("empty too many times, exit");
        }catch (Exception e){
            logger.log(Level.SEVERE, "连接异常"+e);
        }
        finally {
            connector.disconnect();
            logger.log(Level.SEVERE, "断开连接");

        }
    }

    private  void printEntry(List<CanalEntry.Entry> entrys) {
        for (CanalEntry.Entry entry : entrys) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }

            CanalEntry.RowChange rowChage = null;
            try {
                rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
                        e);
            }
            //操作类型
            CanalEntry.EventType operationType = rowChage.getEventType();
            //数据库名
            String databaseName = entry.getHeader().getSchemaName();
            //表名
            String tableName = entry.getHeader().getTableName();

            //过滤操作类型，只记录，增，改
            if(operationType.toString().equals("DELETE")) continue;
            //过滤数据库，只记录test数据库
            if(!databaseName.equals("remote_meter_vlinker")) continue;

            StringBuffer sql = new StringBuffer();

            //遍历内容
            for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
                if (operationType == CanalEntry.EventType.INSERT) {
                    sql.append(" insert into "+tableName +" VALUES ( ");
                    List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                    for (CanalEntry.Column column : afterColumnsList) {
                        String value = column.getValue();
                        if("".equals(value)||value.length()==0){
                            value="0";
                            sql.append("'"+ value +"',");

                        }else {
                            sql.append("'"+ value +"',");
                        }
                    }
                    sql.deleteCharAt(sql.length()-1);
                    sql.append(")");
                    System.out.println(sql.toString());
                    logger.log(Level.INFO, "查询语句 " + sql.toString());

                    mapper.insertMethod(sql.toString());

                } else if(operationType == CanalEntry.EventType.UPDATE){
                    sql.append("update "+ tableName+"  set ");
                    List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                    for (CanalEntry.Column column : afterColumnsList) {

                        //字段有更新
                        if(column.getUpdated()){
                            String s = column.getName()+" = '" +column.getValue() + "' ,";
                            sql.append(s);
                        }
                    }
                    sql.deleteCharAt(sql.length()-1);
                    CanalEntry.Column columnId = afterColumnsList.get(0);
                    sql.append(" where   "+columnId.getName()+"='"+columnId.getValue()+"'");
                    System.out.println(sql.toString());
                    logger.log(Level.INFO, "查询语句 " + sql.toString());

                    mapper.updateMethod(sql.toString());
                }
            }

        }
    }



}
